package cn.edu.xmu.datamining.tangzk.mralgos.parallelbfs;

import java.io.File;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;

import cn.edu.xmu.datamining.tangzk.mralgos.AlgorithmDriver;

/**
 * algorithm for Parallel Breadth First Search
 * 
 * @version 2013-8-1
 * @author tangzk
 * @Reviewer
 * 
 */
public class ParalleBFSDriver extends Configured implements Tool {

	// termination counter
	public static final String ITER_COUNTER = "Need_Iteration";

	@Override
	public int run(String[] args) throws Exception {
		int exitCode = 0;
		if (args.length < 2) {
			throw new Exception("pbfs <inputDir> <outputDir>");
		}
		Configuration conf = this.getConf();
		Job job = null;
		long counter = 1;
		int iteration = 0;

		Path inputDir = new Path(args[0]);
		String baseOutputPath = args[1];

		FileSystem fs = FileSystem.get(conf);
		FileUtil.copy(fs, inputDir, fs, new Path(baseOutputPath + iteration),
				false, true, conf);

		while (counter > 0) {
			job = new Job(conf, "ParallelBFS-" + iteration);
			job.setMapperClass(ParalleBFSMapper.class);
			job.setReducerClass(ParalleBFSReducer.class);
			job.setMapOutputKeyClass(IntWritable.class);
			job.setMapOutputValueClass(GraphNodeWritable.class);
			job.setOutputKeyClass(IntWritable.class);
			job.setOutputValueClass(Text.class);

			FileInputFormat.addInputPath(job, new Path(baseOutputPath
					+ iteration));
			FileOutputFormat.setOutputPath(job, new Path(baseOutputPath
					+ (iteration + 1)));

			if (AlgorithmDriver.runJob(job) < 0) {
				exitCode = -1;
				break;
			}
			counter = job.getCounters()
					.findCounter("UserDefined", ParalleBFSDriver.ITER_COUNTER)
					.getValue();
			iteration++;
		}
		return exitCode;
	}

	public static void main(String[] args) throws Exception {
		AlgorithmDriver.toolRun(new ParalleBFSDriver(), args);
	}

}
